package com.caiya.kafka.listener;

import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import com.caiya.kafka.ConsumerFactory;
import com.caiya.kafka.lifecycle.SmartLifecycle;
import com.caiya.kafka.listener.config.ContainerProperties;
import com.caiya.kafka.util.Assert;
import com.caiya.kafka.util.BeanUtils;
import com.caiya.kafka.util.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The base implementation for the {@link MessageListenerContainer}.
 *
 * @param <K> the key type.
 * @param <V> the value type.
 *
 * @author Gary Russell
 * @author Marius Bogoevici
 * @author Artem Bilan
 */
public abstract class AbstractMessageListenerContainer<K, V>
        implements GenericMessageListenerContainer<K, V> {

    /**
     * The default {@link SmartLifecycle} phase for listener containers {@value #DEFAULT_PHASE}.
     */
    public static final int DEFAULT_PHASE = Integer.MAX_VALUE - 100; // late phase

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    /**
     * The offset commit behavior enumeration.
     */
    public enum AckMode {

        /**
         * Commit after each record is processed by the listener.
         */
        RECORD,

        /**
         * Commit whatever has already been processed before the next poll.
         */
        BATCH,

        /**
         * Commit pending updates after
         * {@link ContainerProperties#setAckTime(long) ackTime} has elapsed.
         */
        TIME,

        /**
         * Commit pending updates after
         * {@link ContainerProperties#setAckCount(int) ackCount} has been
         * exceeded.
         */
        COUNT,

        /**
         * Commit pending updates after
         * {@link ContainerProperties#setAckCount(int) ackCount} has been
         * exceeded or after {@link ContainerProperties#setAckTime(long)
         * ackTime} has elapsed.
         */
        COUNT_TIME,

        /**
         * User takes responsibility for acks using an
         * {@link AcknowledgingMessageListener}.
         */
        MANUAL,

        /**
         * User takes responsibility for acks using an
         * {@link AcknowledgingMessageListener}. The consumer
         * immediately processes the commit.
         */
        MANUAL_IMMEDIATE,

    }

    protected final ConsumerFactory<K, V> consumerFactory; // NOSONAR (final)

    private final ContainerProperties containerProperties;

    private final Object lifecycleMonitor = new Object();

    private String beanName;

    private boolean autoStartup = true;

    private int phase = DEFAULT_PHASE;

    private AfterRollbackProcessor<K, V> afterRollbackProcessor = new DefaultAfterRollbackProcessor<>();

    private volatile boolean running = false;

    private volatile boolean paused;

    /**
     * Construct an instance with the provided properties.
     * @param containerProperties the properties.
     * @deprecated in favor of
     * {@link #AbstractMessageListenerContainer(ConsumerFactory, ContainerProperties)}.
     */
    @Deprecated
    protected AbstractMessageListenerContainer(ContainerProperties containerProperties) {
        this(null, containerProperties);
    }

    /**
     * Construct an instance with the provided factory and properties.
     * @param consumerFactory the factory.
     * @param containerProperties the properties.
     */
    protected AbstractMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                                               ContainerProperties containerProperties) {

        Assert.notNull(containerProperties, "'containerProperties' cannot be null");
        this.consumerFactory = consumerFactory;
        if (containerProperties.getTopics() != null) {
            this.containerProperties = new ContainerProperties(containerProperties.getTopics());
        }
        else if (containerProperties.getTopicPattern() != null) {
            this.containerProperties = new ContainerProperties(containerProperties.getTopicPattern());
        }
        else {
            this.containerProperties = new ContainerProperties(containerProperties.getTopicPartitions());
        }

        BeanUtils.copyProperties(containerProperties, this.containerProperties,
                "topics", "topicPartitions", "topicPattern", "ackCount", "ackTime");

        if (containerProperties.getAckCount() > 0) {
            this.containerProperties.setAckCount(containerProperties.getAckCount());
        }
        if (containerProperties.getAckTime() > 0) {
            this.containerProperties.setAckTime(containerProperties.getAckTime());
        }
        if (this.containerProperties.getConsumerRebalanceListener() == null) {
            this.containerProperties.setConsumerRebalanceListener(createSimpleLoggingConsumerRebalanceListener());
        }
        if (containerProperties.getGenericErrorHandler() instanceof BatchErrorHandler) {
            this.containerProperties.setBatchErrorHandler((BatchErrorHandler) containerProperties.getGenericErrorHandler());
        }
        else {
            this.containerProperties.setErrorHandler((ErrorHandler) containerProperties.getGenericErrorHandler());
        }
    }

    @Override
    public boolean isAutoStartup() {
        return this.autoStartup;
    }

    public void setAutoStartup(boolean autoStartup) {
        this.autoStartup = autoStartup;
    }

    protected void setRunning(boolean running) {
        this.running = running;
    }

    @Override
    public boolean isRunning() {
        return this.running;
    }

    protected boolean isPaused() {
        return this.paused;
    }

    @Override
    public boolean isPauseRequested() {
        return this.paused;
    }

    public void setPhase(int phase) {
        this.phase = phase;
    }

    @Override
    public int getPhase() {
        return this.phase;
    }

    protected AfterRollbackProcessor<K, V> getAfterRollbackProcessor() {
        return this.afterRollbackProcessor;
    }

    /**
     * Set a processor to perform seeks on unprocessed records after a rollback.
     * Default will seek to current position all topics/partitions, including the failed
     * record.
     * @param afterRollbackProcessor the processor.
     * @since 1.3.5
     */
    public void setAfterRollbackProcessor(AfterRollbackProcessor<K, V> afterRollbackProcessor) {
        Assert.notNull(afterRollbackProcessor, "'afterRollbackProcessor' cannot be null");
        this.afterRollbackProcessor = afterRollbackProcessor;
    }

    @Override
    public ContainerProperties getContainerProperties() {
        return this.containerProperties;
    }

    @Override
    public void setupMessageListener(Object messageListener) {
        this.containerProperties.setMessageListener(messageListener);
    }

    @Override
    public final void start() {
        checkGroupId();
        synchronized (this.lifecycleMonitor) {
            if (!isRunning()) {
                Assert.isTrue(
                        this.containerProperties.getMessageListener() instanceof GenericMessageListener,
                        "A " + GenericMessageListener.class.getName() + " implementation must be provided");
                doStart();
            }
        }
    }

    public void checkGroupId() {
        if (this.containerProperties.getTopicPartitions() == null) {
            boolean hasGroupIdConsumerConfig = true; // assume true for non-standard containers
            if (this.consumerFactory != null) { // we always have one for standard containers
                Object groupIdConfig = this.consumerFactory.getConfigurationProperties()
                        .get(ConsumerConfig.GROUP_ID_CONFIG);
                hasGroupIdConsumerConfig = groupIdConfig != null && groupIdConfig instanceof String
                        && StringUtils.hasText((String) groupIdConfig);
            }
            Assert.state(hasGroupIdConsumerConfig || StringUtils.hasText(this.containerProperties.getGroupId()),
                    "No group.id found in consumer config, container properties, or @KafkaListener annotation; "
                            + "a group.id is required when group management is used.");
        }
    }

    protected abstract void doStart();

    @Override
    public final void stop() {
        synchronized (this.lifecycleMonitor) {
            if (isRunning()) {
                final CountDownLatch latch = new CountDownLatch(1);
                doStop(latch::countDown);
                try {
                    latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS);
                }
                catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    @Override
    public void pause() {
        this.paused = true;
    }

    @Override
    public void resume() {
        this.paused = false;
    }

    @Override
    public void stop(Runnable callback) {
        synchronized (this.lifecycleMonitor) {
            if (isRunning()) {
                doStop(callback);
            }
        }
    }

    protected abstract void doStop(Runnable callback);

    /**
     * Return default implementation of {@link ConsumerRebalanceListener} instance.
     * @return the {@link ConsumerRebalanceListener} currently assigned to this container.
     */
    protected final ConsumerRebalanceListener createSimpleLoggingConsumerRebalanceListener() {
        return new ConsumerRebalanceListener() {

            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                AbstractMessageListenerContainer.this.logger.info("partitions revoked: " + partitions);
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                AbstractMessageListenerContainer.this.logger.info("partitions assigned: " + partitions);
            }

        };
    }

    public void setBeanName(String name) {
        this.beanName = name;
    }

    public String getBeanName() {
        return this.beanName;
    }

}
